钉钉python实现审批（通过软件和钉钉审批流的结合） |
您所在的位置：网站首页 › 钉钉 工贼 › 钉钉python实现审批（通过软件和钉钉审批流的结合） |
钉钉python实现审批（通过软件和钉钉审批流的结合）
|
为了实现工单系统，又不想自己去结合钉钉的组织架构实现一套审批流，所以采用钉钉的审批结合自己的系统去实现工单。 由钉钉实现审批流，软件实现查询，记录工单等。 一、具体流程
二、具体的实现方式 获取token的函数 def getToken(self): ''' 获取钉钉的token :return: 钉钉token ''' url = 'https://oapi.dingtalk.com/gettoken?appkey=XXXXXXXXXXXXXXXXX&appsecret=XXXXXXXXXXXXXXXXXXXXXXXXXX' req = requests.get(url) req = json.loads(req.text) return req['access_token']1. 注册回调函数 def createCallbackDd(self, request): ''' 注册钉钉回调函数 :return: ''' url = 'https://oapi.dingtalk.com/call_back/register_call_back?access_token=' + self.getToken() data = { "call_back_tag": ["bpms_task_change", "bpms_instance_change"], #这两个回调种类是审批的 "token": TOKEN, #自定义的字符串 "aes_key": AES_KEY, #自定义的43位字符串，密钥 "url": URL #回调地址 } requests.post(url, data=json.dumps(data)) return HttpResponse('OK')
2. 创建钉钉审批 注意：不使用python的dingtalk-sdk，这个版本必须要传approvers（审批人）这个参数，导致不能使用后台配置的工单流，直接使用接口或者找到最新版的sdk def create(self, request): ''' 创建钉钉审批 :param request: :return: ''' url = 'https://oapi.dingtalk.com/topapi/processinstance/create?access_token=' + self.getToken() form_component_value_vo = [{'name': '姓名', 'value': '测试2'}, {'name': '部门', 'value': '测试2'}, {'name': '["开始时间","结束时间"]', 'value': '["2019-06-10 10:32","2019-06-10 10:36"]'}, {'name': '加班事由', 'value': '测试2'}] #钉钉后台配置的需要填写的字段（这一段由点到传过来实现，这里只是测试使用） data = { 'agent_id': None, 'process_code': 'XXXXXXXXXXXXXXXX', #工单id 'originator_user_id': 'manager2606', #创建人的钉钉userid 'dept_id': -1, #创建人的钉钉部门id 'form_component_values': str(form_component_value_vo), #钉钉后台配置的需要填写的字段 } response = requests.post(url, data=data) print(response.text) return HttpResponse('成功')
3. 回调函数 在您注册事件回调接口的时候，钉钉服务器会向您“注册回调接口”时候上传的url(接收回调的url)推送一条消息，用来测试url的合法性。收到消息后，需要返回经过加密后的字符串“success”的json数据，否则钉钉服务器将认为url不合法。 class Crypto(object): def __init__(self): # 随便填的字符串 self.token = TOKEN # 自己生成的43位随机字符串 self.aes_key = AES_KEY # 钉钉企业ID self.corpid = CORPID self.crypto = DingTalkCrypto( token=self.token, encoding_aes_key=self.aes_key, corpid_or_suitekey=self.corpid ) def callbackDd(self, request): encrypt = request.data.get('encrypt') msgSignature = request.GET.get("signature") timeStamp = request.GET.get("timestamp") nonce = request.GET.get("nonce") callback = json.loads(Crypto().crypto.decrypt_encrypt_str( encrypt_str=encrypt, nonce=nonce, timestamp=timeStamp, signature=msgSignature )) print(callback) #第一次调用回调函数的验证，关键字是check_url if callback['EventType'] == 'check_url': result = Crypto().crypto.encrypt_message( msg="success", nonce="123456", timestamp=int(time.time() * 1000) ) return Response(result) # bpms_task_change : 审批任务开始，结束，转交 # bpms_instance_change：审批实例开始，结束 elif callback['EventType'] == 'bpms_task_change': pass #实现入库 elif callback['EventType'] == 'bpms_instance_change': pass #实现入库
三、钉钉的后台配置 参考钉钉的文档：https://open-doc.dingtalk.com/microapp/serverapi2/tvu5f4 上面的代码实现，我的配置方式是：
结尾：后续的数据库设计会在后面的内容中展示。 |
【本文地址】